package com.shujia.core

import org.apache.flink.streaming.api.scala._

object Demo02RPCSubmitJob {
  def main(args: Array[String]): Unit = {
    // 通过RPC远程调用方式提交任务
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createRemoteEnvironment(
      "node1"
      , 44867
      , "C:\\Users\\zzk10\\IdeaProjects\\Bigdata16\\Flink\\target\\Flink-1.0.jar")
    // 默认并行度等于CPU的逻辑核数 相当于是任务的一个并行度
    env.setParallelism(2)

    /**
     * 通过Socket模拟实时数据
     * nc -lk 8888
     *
     * DataStream: Flink中的编程模型 类似Spark中的DStream
     * 通过Socket创建一个无界流
     */
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //    linesDS.print()

    // 对每一条数据进行切分
    val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(","))

    // 将每个单词变成 K V格式
    val wordsKVDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))

    // 按照每个单词进行分组
    val keyByDS: KeyedStream[(String, Int), String] = wordsKVDS.keyBy(kv => kv._1)

    // 统计每个单词的数量 指定位置进行累加
    val wordCntDS: DataStream[(String, Int)] = keyByDS.sum(1)

    /**
     * Flink中的算子自带状态
     */
    // 将结果打印
    wordCntDS.print()

    // 启动任务
    env.execute("Demo02RPCSubmitJob")

  }

}
